package com.kafka.consumer;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
import net.sf.json.JSONObject;

import com.kafka.core.BaseClass;
import com.kafka.util.HttpUtil;

public class PayConsumer extends BaseClass implements Runnable {

	private final ConsumerConnector consumer;

	private static String TOPIC;
	private static String syncUri;
	private static String payUri;
//	private static String rechargeUri;
	private final static Integer PAY = 1;
	private final static Integer SYNC = 2;

	public PayConsumer() {
		Properties props = new Properties();
		try {
			props.load(ClassLoader.getSystemResourceAsStream("pay.properties"));
			TOPIC = props.getProperty("topic");
			syncUri = props.getProperty("sync_uri");
			payUri = props.getProperty("pay_uri");
//			rechargeUri = props.getProperty("recharge_uri");
		} catch (IOException e) {
			e.printStackTrace();
		}
		ConsumerConfig config = new ConsumerConfig(props);
		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
	}

	public void consume() throws IOException {
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(TOPIC, new Integer(1));

		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
		StringDecoder valueDecoder = new StringDecoder(
				new VerifiableProperties());

		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer
				.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
		KafkaStream<String, String> stream = consumerMap.get(TOPIC).get(0);
		ConsumerIterator<String, String> it = stream.iterator();
		// 遍历KafkaStream
		while (it.hasNext()) {
			// 获得MessageAndMetadata对象
			MessageAndMetadata<String, String> mam = it.next();
			// JSONObject rs = null;
			try {
				if(!mam.message().isEmpty()){
//					String jsonStr = new String(mam.message().getBytes(), encode);
					String jsonStr = mam.message();
					JSONObject json = rebuildJSON(JSONObject.fromObject(jsonStr));
					if(json.getInt("api")==SYNC){
						HttpUtil.doPost(syncUri, json.getString("datas"),json.getString("appid"),json.getString("secretkey"));
					}else if(json.getInt("api")==PAY){
						HttpUtil.doPost(payUri, json.getString("datas"),json.getString("appid"),json.getString("secretkey"));
					}
//					else if(MsgType.RECHARGE.toString().equals(json.get("msgtype").toString())){
//						HttpUtil.doPost(syncUri, json.get("datas").toString(),appID,secretKey);
//					}
					// log.info("###RegisterConsumer:"+json.get("datas").toString()+",result:"+rs);
					addMessage(mam, success, "Send Message successful！");
				}else{
					addMessage(mam, fail, "message is empty!");
				}
			} catch (Exception e) {
				// log.info("###RegisterConsumer:"+e.getMessage()+",result:"+rs);
				addMessage(mam, fail, e.toString());
			}
		}
	}

	@Override
	public void run() {
		try {
			consume();
		} catch (UnsupportedEncodingException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
//public class PayConsumer {
//
//	private final ConsumerConnector consumer;
//	private ExecutorService executor;
//
//	private static String TOPIC;
//	private static String ENCODE;
//	private static String postUri;
//
//	public PayConsumer() {
//		Properties props = new Properties();
//		try {
//			props.load(ClassLoader
//					.getSystemResourceAsStream("pay.properties"));
//			TOPIC = props.getProperty("topic");
//			ENCODE = props.getProperty("encode");
//			postUri = props.getProperty("post_uri");
//		} catch (IOException e) {
//			e.printStackTrace();
//		}
//		ConsumerConfig config = new ConsumerConfig(props);
//		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
//	}
//
//    public void shutdown() {
//        if (consumer != null)
//            consumer.shutdown();
//        if (executor != null)
//            executor.shutdown();
//    }
//
//	public void consume(int numThreads) throws IOException {
//		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
//		topicCountMap.put(TOPIC, new Integer(numThreads));
//
//		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
//		StringDecoder valueDecoder = new StringDecoder(
//				new VerifiableProperties());
//
//		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer
//				.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
//		List<KafkaStream<String, String>> streams = consumerMap.get(TOPIC);
//		
//        // now launch all the threads
//        executor = Executors.newFixedThreadPool(numThreads);
// 
//        // now create an object to consume the messages
//        //
//        int threadNumber = 0;
//        for (final KafkaStream stream : streams) {
//            executor.submit(new ConsumerMsgTask(stream, threadNumber,postUri,ENCODE));
//            threadNumber++;
//        }
//	}
//}